package com.lagou.rpc.consumer.proxy;

import com.lagou.rpc.consumer.client.RpcClient;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.*;

public class RpcClientFactory {

    private static RpcClientFactory rpcClientFactory;

    private   ZkClient zkClient;

    private  String serversDir = "/servers";

    private  HashMap<String, RpcClient> servers = new HashMap<>();

    private RpcClientFactory() {

        zkClient = new ZkClient("127.0.0.1:2181");
        System.out.println("会话被创建了...");

        List<String> children = new ArrayList<>();

        try {
            children = zkClient.getChildren(serversDir);
        } catch (Exception e) {

            e.printStackTrace();
        }

        initRpcClients(children);

        zkClient.subscribeChildChanges(serversDir, new IZkChildListener() {
            @Override
            public void handleChildChange(String s, List<String> list) throws Exception {
                System.out.println(s + " : " + list);
                for (String address : list) {
                    if (!servers.containsKey(address)) {
                        addRpcClient(address);
                    }
                }
            }
        });

        Timer timer = new Timer();

        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                dataReport();
            }
        }, 0, 5000);

    }


    private void dataReport() {
        if (servers != null) {
            for (String address : servers.keySet()) {
                RpcClient rpcClient = servers.get(address);
                String nodePath = serversDir + "/" + address;

                String content = (String)zkClient.readData(nodePath);

                int interval = (int)(System.currentTimeMillis() - rpcClient.getLastResponseTime());


                if (interval > 5000) {
                    zkClient.writeData(nodePath, null);
                    rpcClient.setLastResponseTime(0);
                    rpcClient.setLastResponseCost(0);
                } else {
//                    System.out.println("-- 上报： 耗时：" + rpcClient.getLastResponseCost() + "时间：" + rpcClient.getLastResponseTime());
                    zkClient.writeData(nodePath, rpcClient.getLastResponseCost() + ":" + rpcClient.getLastResponseTime());
                }

            }
        }
    }

    public static RpcClientFactory getInstance() {
        if (rpcClientFactory == null) {
            rpcClientFactory = new RpcClientFactory();
        }
        return rpcClientFactory;
    }



    public  void addRpcClient(String address) {
        System.out.println("------- " + address + " online ------");

        // 解析IP和端口
        String[] split = address.split(":");
        String ip = split[0];
        int port = Integer.parseInt(split[1]);

        servers.put(address, new RpcClient(ip, port));

        zkClient.subscribeDataChanges(serversDir + "/" + address, new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
//                System.out.println(s + " changed:" + o);
            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
//                System.out.println(s + " deleted:");
                String[] split = s.split("/");
                String address = split[split.length - 1];

                System.out.println("------- " + address + " offline ------");
                servers.remove(address);
            }
        });
    }

    public  void initRpcClients(List<String> serverAddress) {
        for (String address : serverAddress) {
            addRpcClient(address);
        }

    }

    private int serverIndex = 0;


    // 轮流负载建立rpc连接
    public RpcClient createRpcClient() {

        Set<String> minResponseAddresses = new HashSet<>();
        long minResponseCost = Long.MAX_VALUE;

        // 取最后一次响应时间最小的服务器作为本次负载
        for (String address : servers.keySet()) {
            if (servers.get(address).getLastResponseCost() < minResponseCost) {
                minResponseAddresses = new HashSet<String>(){{
                    add(address);
                }};
                minResponseCost = servers.get(address).getLastResponseCost();
            } else if (servers.get(address).getLastResponseCost() == minResponseCost) {
                minResponseAddresses.add(address);
            }
        }

        if (minResponseAddresses.size() == 0) {
            return null;
        } else {    // 当有多个相同最小值时，随机选取一个服务器作为负载
            List<String> list = new ArrayList<>(minResponseAddresses);
            String address = list.get((int) (list.size() * Math.random()));
            return servers.get(address);
        }
    }
}
